﻿using LucyGrid.Interface;
using LucyGrid.Master;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;
using LucyGrid.Messaging;

namespace LucyGrid
{
	/// <summary>
	/// .NET Grid implementation
	/// Based on NGrid and Alchemy project allow to distribute computation across computers.
	/// <remarks>GridMaster setups an endpoit and requests for grid computer across the network
	/// sending an event over a multicast request.
	/// Lucy: Because we use the 10% computing power available at the office.</remarks>
	/// </summary>
	public class GridMaster : ILucyMaster, IDisposable
	{
		byte[] conf;
		GridEndpoint server;
		ChunkBucket chunkBucket;
		bool endGrid = false;
		GridComm bus;
		GridHeartbeat heartbeat;
		GridSettings settings;
		private GridInfo gridInfo;

		#region EVENTS
		/// <summary>
		/// New node detected
		/// </summary>
		/// <param name="node"></param>
		public delegate void NewNodeHandler(NodeInfo node);
		/// <summary>
		/// A node hearbeat has timeout
		/// </summary>
		/// <param name="node"></param>
		public delegate void NodeTimeout(NodeInfo node);
		/// <summary>
		/// A new chunk has been created
		/// </summary>
		/// <param name="chunk"></param>
		public delegate void ChunkCreatedHandler(ChunkInfo chunk);
		/// <summary>
		/// Chunk has been sent to a node for processing
		/// </summary>
		/// <param name="chunk"></param>
		public delegate void ChunkRequestHandler(ChunkInfo chunk);
		/// <summary>
		/// A chunk process failed. Chunk will be released and ready to be processed by another node
		/// </summary>
		/// <param name="chunk"></param>
		public delegate void ChunkResetHandler(ChunkInfo chunk, string reason);
		/// <summary>
		/// Chunk processed and ready for deliver
		/// </summary>
		/// <param name="chunk"></param>
		public delegate void ChunkProcessEndHandler(ChunkInfo chunk);

		/// <summary>
		/// New node detected
		/// </summary>
		/// <param name="node"></param>
		public event NewNodeHandler OnNewNode;
		/// <summary>
		/// A node hearbeat has timeout
		/// </summary>
		/// <param name="node"></param>
		public event NodeTimeout OnNodeTimeout;
		/// <summary>
		/// A new chunk has been created
		/// </summary>
		/// <param name="chunk"></param>
		public event ChunkCreatedHandler OnChunkCreated;
		/// <summary>
		/// Chunk has been sent to a node for processing
		/// </summary>
		/// <param name="chunk"></param>
		public event ChunkRequestHandler OnChunkRequest;
		/// <summary>
		/// A chunk process failed. Chunk will be released and ready to be processed by another node
		/// </summary>
		/// <param name="chunk"></param>
		public event ChunkResetHandler OnChunkReset;
		/// <summary>
		/// Chunk processed and ready for deliver
		/// </summary>
		/// <param name="chunk"></param>
		public event ChunkProcessEndHandler OnChunkProcessEnd;


		#endregion EVENTS

		/// <summary>
		/// Remote processing request Type
		/// <remarks>The type of object to serialize for input data</remarks>
		/// </summary>
		Type requestType;
		/// <summary>
		/// Remote processing response Type
		/// <remarks>The type of object to deserialize from the output data</remarks>
		/// </summary>
		Type responseType;


		static GridMaster()
		{
			GridUtils.Mode = GridUtils.ActorType.MASTER;
		}
		/// <summary>
		/// Build one grid for each method you need to distribute
		/// </summary>
		/// <param name="mi"></param>
		public GridMaster(MethodInfo mi, GridSettings customSettings)
		{
			this.settings = customSettings ?? new GridSettings();
			bus = new GridComm();
			conf = CreateConfiguration(mi);
			chunkBucket = new ChunkBucket();
			requestType = mi.GetParameters().First().ParameterType;
			responseType = mi.ReturnType;
			var id = GenerateId();
			gridInfo = new GridInfo()
			{
				Tags = "GRID,MASTER",
				Id = id,
				Endpoint = new Uri(string.Format("http://{0}:{1}/{2}/", GridComm.ResolveName(Environment.MachineName), GetFreeTcpPort(), id)),
				Chunks = 0,
				EnableLog = GridUtils.EnableLog,
				Metrics = chunkBucket.Metrics,
				Stamp = DateTime.Now
			};
			server = new GridEndpoint(gridInfo.Endpoint.ToString());
			server.RequestReceived += server_RequestReceived;
			heartbeat = new GridHeartbeat(gridInfo.Endpoint.ToString(), settings.ChunkTimeout);
			heartbeat.OnNodeTimeout += (node) =>
			{
				if (OnNodeTimeout != null)
					OnNodeTimeout(node);
				PurgeChunks(node);
			};
			// Ready, go!
			server.Start();
			GridUtils.NotifyLog("MASTER,INFO", "GridMaster initialized. Enpoint {0}", gridInfo.Endpoint);
			// Run local processing if it's not deactivated
			if (settings.EnableLocalProcessing)
				RunLocalExecutor(mi);
		}

		private void RunLocalExecutor(MethodInfo mi)
		{
			GridUtils.NotifyLog("MASTER,INFO", "Running LOCAL EXECUTOR");
			ILucyMaster master = this;
			Task.Run(() =>
			{
				Object caller = LucyBinManager.CreateInstance(mi.DeclaringType);
				new GridExecutor(caller, mi, gridInfo.Endpoint.ToString(), master);
			});
		}

		/// <summary>
		/// Creates configuration section
		/// </summary>
		/// <param name="mi"></param>
		/// <returns></returns>
		private byte[] CreateConfiguration(MethodInfo mi)
		{
			return UTF8Encoding.UTF8.GetBytes(new XElement("conf",
					new XElement[]{
						new XElement("files", 
							GetBinFiles(mi)),
						new XElement("exetype", mi.DeclaringType.FullName),
						new XElement("exemethod", mi.ToString())
					}).ToString());
		}


		/// <summary>
		/// Returns all the binary files to be distributed to the nodes
		/// </summary>
		/// <returns></returns>
		private IEnumerable<XElement> GetBinFiles(MethodInfo mi)
		{
			var mainAssemblyLocation = Path.GetDirectoryName(mi.DeclaringType.Assembly.Location);
			if (settings.BinPath == null)
				settings.BinPath = new List<string>();
			settings.BinPath.Add(mainAssemblyLocation);
			return LucyBinManager.GetBinaries(settings.BinPath).Keys.Select(f => new XElement("file", f));
		}

		/// <summary>
		/// Notifies grid request
		/// </summary>
		/// <remarks>Each notfication indicates the number of chunks left</remarks>
		protected void NotifyGrid()
		{
			GridMaster grid = this;
			Task.Run(() =>
			{
				Thread.CurrentThread.Name = string.Format("GRIDMASTER NOTIFIER-{0}", Thread.CurrentThread.ManagedThreadId);
				while (!grid.endGrid)
				{
					EmitGridChunksInfo();
					Thread.Sleep(1000);
				}
			});
		}

		/// <summary>
		/// Notifies about grid neeeds
		/// </summary>
		private void EmitGridChunksInfo()
		{
			gridInfo.Stamp = DateTime.Now;
			gridInfo.Chunks = chunkBucket.InButcketCount;
			bus.Emit(gridInfo.Tags, GridCommEvents.grid.ToString(), gridInfo);
		}


		/// <summary>
		/// Node down. Reset all chunks associated to this node
		/// </summary>
		private void PurgeChunks(NodeInfo node)
		{
			GridUtils.NotifyLog("MASTER,INFO", "Releasing chunks locked by: {0}", node.Id);
			chunkBucket.ResetChunkProcessing(node);
		}

		/// <summary>
		/// Generate an GUID for the grid
		/// </summary>
		/// <returns></returns>
		private string GenerateId()
		{
			return Guid.NewGuid().ToString().Replace("-", string.Empty);
		}

		/// <summary>
		/// Process grid nodes requests
		/// </summary>
		/// <param name="sender"></param>
		/// <param name="context"></param>
		void server_RequestReceived(object sender, HttpListenerContext context)
		{
			var url = context.Request.Url.PathAndQuery;
			GridUtils.NotifyLog("MASTER,INFO", "Processing request: {0}", url);
			var error = false;
			var node = GetNodeInfo(context.Request);
			// Settings
			if (url.Contains("/settings"))
			{
				context.Response.Headers.Add("Content-Type", "text/xml");
				using (var os = context.Response.OutputStream)
				{
					os.Write(conf, 0, conf.Length);
				}
			}
			// File download
			else if (url.Contains("/file"))
			{
				using (var os = context.Response.OutputStream)
				{
					var index = url.IndexOf("/file") + 5;
					var file = url.Substring(index + 1);
					context.Response.Headers.Add("Content-Type", "application/octet-stream");
					try
					{
						var data = LucyBinManager.GetAssemblyBinary(file);
						os.Write(data, 0, data.Length);
					}
					catch (Exception ex)
					{
						GridUtils.NotifyLog("MASTER,INFO", "Error loading resource file {0}: {1}", file, ex.ToString());
						context.Response.StatusCode = (int)HttpStatusCode.NotFound;
					}
				}
			}
			// Missing assembly
			else if (url.Contains("/assembly/"))
			{
				using (var os = context.Response.OutputStream)
				{
					var fullname = url.Split('/').Last();
					try
					{
						byte[] assmdata = LucyBinManager.GetAssemblyBinary(fullname);
						os.Write(assmdata, 0, assmdata.Length);
					}
					catch (Exception ex)
					{
						context.Response.StatusCode = 500;
						GridUtils.NotifyLog("MASTER,INFO", "Error loading assembly: {0}", ex.ToString());
					}
				}
			}
			// Chunk RES
			else if (url.Contains("/chunk/error/"))
			{
				var errorMessage = context.Request.Headers["error"];
				using (var os = context.Response.OutputStream)
				{
					var chunkid = int.Parse(url.Split('/').Last());
					ResetChunkProcessing(chunkid, string.Format("Chunk ERROR response {0}", errorMessage));
				}
			}
			// Chunk REQ/RES
			else if (url.Contains("/chunk"))
			{
				if (context.Request.HttpMethod.Contains("POST"))
				{
					var chunkid = int.Parse(url.Split('/').Last());
					try
					{
						// Receive response
						using (var inputsream = context.Request.InputStream)
						{
							var processedChunk = new Chunk(inputsream, responseType);
							FinalizeChunk(processedChunk);
						}
					}
					catch (Exception ex)
					{
						GridUtils.NotifyLog("MASTER,INFO", "Chunk request error: " + ex.Message);
						error = true;
						ResetChunkProcessing(chunkid, ex.Message);
					}
				}
				// Return next chunk
				if (context.Request.HttpMethod.Contains("GET"))
				{
					if (!ServerReturnNextChunk(context, node))
						error = true;
				}
			}
			else
			{
				using (var ostream = context.Response.OutputStream)
				{
				}
			}
			if (error)
				context.Response.StatusCode = 200;
			context.Response.Close();
		}

		/// <summary>
		/// Returns next available chunk
		/// </summary>
		/// <param name="context"></param>
		/// <param name="node"></param>
		private bool ServerReturnNextChunk(HttpListenerContext context, NodeInfo node)
		{
			using (var os = context.Response.OutputStream)
			{
				ChunkInfo chunk = null;
				try
				{
					chunk = GetNextChunk(node);
					if (chunk != null)
					{
						context.Response.Headers.Add("id", chunk.Id.ToString());
						chunk.Chunk.Serialize(os);
					}
					else
						context.Response.StatusCode = (int)HttpStatusCode.NoContent;
				}
				catch (Exception ex)
				{
					GridUtils.NotifyLog("MASTER,INFO", "Chunk request error: " + ex.Message);
					if (chunk != null)
						ResetChunkProcessing(chunk.Id, string.Format("Chunk REQ exception {0}", ex.Message));
					return false;
				}
				if (chunk != null)
				{
					// Set processing
					chunk.Chunk.Trace("SetChunkProcessing");
					GridUtils.NotifyLog("MASTER,CHUNK,INFO", ", {0}, STEP, 2, Chunk processing, {1}", chunk.Id, node.Id);
				}
			}
			return true;
		}

		/// <summary>
		/// Retuns information about the node connection to the master
		/// </summary>
		/// <param name="httpListenerRequest"></param>
		/// <returns></returns>
		private NodeInfo GetNodeInfo(HttpListenerRequest request)
		{
			var id = request.Headers["id"];
			var newnode = !heartbeat.Nodes.Any(n => n.Id == id);
			var node = heartbeat.RegisterNode(id, request.RemoteEndPoint.Address.ToString());
			if (newnode && OnNewNode != null)
				OnNewNode(node);
			return node;
		}

		/// <summary>
		/// Adds a new chunk to be processed
		/// </summary>
		/// <param name="id"></param>
		/// <param name="data"></param>
		/// <returns></returns>
		private ChunkInfo CreateChunk(int id, object data)
		{
			var chunk = chunkBucket.Add(id, data);
			if (OnChunkCreated != null)
				OnChunkCreated(chunk);
			return chunk;
		}

		/// <summary>
		/// Mark chunk as finalized and ready to be returned to the main process
		/// </summary>
		/// <param name="chunkid">Chunk identifier</param>
		/// <param name="data">Data generated by the grid node</param>
		private void FinalizeChunk(Chunk processedChunk)
		{
			var chunk =  chunkBucket.FinalizeChunk(processedChunk);
			GridUtils.NotifyLog("MASTER,INFO", "Finalize chunk {0}, process time: {1} sec.", chunk.Id, (DateTime.Now - chunk.StartProcessing).Seconds);
			if (OnChunkProcessEnd != null)
				OnChunkProcessEnd(chunk);
		}

		/// <summary>
		/// Error processing a chunk. Reset flags to be ready to be processed by another node.
		/// </summary>
		/// <param name="chunkid"></param>
		private void ResetChunkProcessing(int chunkid, string reason)
		{
			var chunk = chunkBucket.ResetChunkProcessing(chunkid);
			// Notify before resetting node info
			if (OnChunkReset != null)
				OnChunkReset(chunk, reason);
			GridUtils.NotifyLog("MASTER,CHUNK,INFO", ", {0}, STEP, 1.1, Reset chunk: {1}", chunk.Id, reason);
		}

		/// <summary>
		/// Get next chunk to process.
		/// <remarks>If all chunks are beign processed, sleep until:
		/// a) all are relased or 
		/// b) there's a new chunk to process</remarks>
		/// </summary>
		/// <returns></returns>
		private ChunkInfo GetNextChunk(NodeInfo ninfo)
		{
			var chunk = chunkBucket.GetNextChunk(ninfo);
			if (chunk != null)
			{
				if (OnChunkRequest != null)
					OnChunkRequest(chunk);
				GridUtils.NotifyLog("MASTER,INFO", "Processing chunk: {0}", chunk.Id);
			}
			return chunk;
		}

		/// <summary>
		/// Get a free TCP port to setup the endpoit
		/// </summary>
		/// <returns></returns>
		public static int GetFreeTcpPort()
		{
#if DEBUG
			return 23107;
#else
			TcpListener l = new TcpListener(IPAddress.Loopback, 0);
			l.Start();
			int port = ((IPEndPoint)l.LocalEndpoint).Port;
			l.Stop();
			return port;
#endif
		}

		/// <summary>
		/// Start processing chunks
		/// </summary>
		/// <remarks>Converts each input into a chunk and returns processed data</remarks>
		/// <param name="inputData">Data to be procesed by the grid</param>
		/// <returns></returns>
		public IEnumerable<TResult> ProcessData<TSource, TResult>(IEnumerable<TSource> inputData)
		{
			try
			{
				GridUtils.NotifyLog("MASTER,INFO", "Grid processing start. Create chunks");
				// Read in background
				StreamIn(inputData);
				// Wait for the first chunks and Start sending notifications
				Thread.Sleep(200);
				NotifyGrid();
				// Return chunks as soon as they are processed
				foreach (var o in StreamOut())
					yield return (TResult)o;
			}
			finally
			{
				GridUtils.NotifyLog("MASTER,INFO", "Grid processing end.");
				// End
				FinalizeGrid();
			}

		}
		/// <summary>
		/// Outputs processed data
		/// </summary>
		private IEnumerable<Object> StreamOut()
		{
			ChunkInfo processed = null;
			while ((processed = chunkBucket.Take()) != null)
			{
				if (processed.Chunk.Head.IsError)
					throw new Exception("One or more errors occurs while processing chunks. To avoid this exception surround your code by a try/catch block.",
						processed.ExceptionError);
				yield return processed.Chunk.Data;
				processed.Chunk.Trace("StreamOut");
				GridUtils.NotifyLog("MASTER,CHUNK,DEBUG", ", {0}, STEP, 4, Stream out", processed.Id);
				GridUtils.NotifyLog("MASTER,CHUNK,DEBUG,DUMP", processed.ToString());
			}
		}

		/// <summary>
		/// Inputs data to be processed by the grid
		/// </summary>
		/// <param name="inputData"></param>
		private void StreamIn<TSource>(IEnumerable<TSource> inputData)
		{
			Task.Run(() =>
			{
				Thread.CurrentThread.Name = string.Format("GRIDENDPOINT STREAMIN-{0}", Thread.CurrentThread.ManagedThreadId);
				try
				{
					var id = 0;
					ChunkInfo chunk = null;
					foreach (var o in inputData)
					{
						chunk = CreateChunk(id++, (object)o);
						GridUtils.NotifyLog("MASTER,CHUNK,INFO", ", {0}, STEP, 1, New chunk", chunk.Id);
						// Notify asap
						if (id == 1)
							EmitGridChunksInfo();
					}
				}
				catch (Exception ex)
				{
					// Notify about the error
					GridUtils.NotifyLog("MASTER,ERROR", "Error processing data: {0}", ex);
				}
				chunkBucket.CompleteAdding();
			});
		}

		/// <summary>
		/// Yes, I'm declaring some kind of a destructor in background...
		/// </summary>
		private void FinalizeGrid()
		{
			Task.Run(() =>
			{
				Thread.CurrentThread.Name = string.Format("GRIDMASTER FINALIZE-{0}", Thread.CurrentThread.ManagedThreadId);
				try
				{
					endGrid = true;
					// Notify 0 chunks left
					EmitGridChunksInfo();
					GridUtils.NotifyLog("MASTER,INFO", "End processing chunks");
					GridUtils.NotifyLog("MASTER,INFO", "Stopping grid server");
					server.Stop();
				}
				catch (Exception ex)
				{
					GridUtils.NotifyLog("MASTER", "ERROR", "Stopping grid server: {0}", ex);
				}
			});
		}

		/// <summary>
		/// Release the grid
		/// </summary>
		public void Dispose()
		{
		}

		public byte[] Assembly(string fullname)
		{
			return LucyBinManager.GetAssemblyBinary(fullname);
		}
	}
}
